package org.niit.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/*

  DStream transform方法： 将 DStream 转换成 RDD 。就可以使用RDD的所有方法对数据进行操作
    为什么要将DStream转换RDD ? 因为DStream的功能不完整

    Spark Streaming 主要功能并不是分析数据，而是可以接收流式数据
    Spark SQL 主要的功能是读取结构化和半结构化数据
    Spark Core 主要功能才是分析
 */
object SparkStreaming_04 {
  def main(args: Array[String]): Unit = {
     //准备环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    ssc.sparkContext.setLogLevel("ERROR")

    //监听本地套接字，端口号 9999
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

    val resDS: DStream[(String, Int)] = lines.transform(rdd => { //DStream --> RDD --> DStream
      //在这里面，DStream 已经转换成RDD ，所以在这里可以使用RDD的所有方法
      val flatMap: RDD[String] = rdd.flatMap(_.split(" "))
      val map: RDD[(String, Int)] = flatMap.map((_, 1))
      val res: RDD[(String, Int)] = map.reduceByKey(_ + _)
      //transform必须要有返回值，并且该值一定是RDD类型
      res
    })

    resDS.print()

    ssc.start()
    ssc.awaitTermination()


  }
}
